from pyspark.sql import SparkSession, functions as F
from pyspark.sql.types import StructType, IntegerType, StringType

if __name__ == '__main__':
    # 构建SparkSession对象
    spark = SparkSession.builder. \
        appName("local[*]"). \
        config("spark.sql.shuffle.partitions", "4"). \
        getOrCreate()
    # appName 设置程序名称
    # config 设置常用属性。可以通过此来设置配置
    # 最后通过getOrCreate 创建 SparkSession对象

    # 从SparkSession中获取SparkContext
    sc = spark.sparkContext

    # TODO 1: 使用SQL风格处理WordCount
    # 以RDD为基础做数据加载
    rdd = sc.textFile("hdfs://node1:8020/input/wordcount/words.txt") \
        .flatMap(lambda line: line.split(' ')) \
        .map(lambda word: [word])

    # RDD转换成DataFrame
    df = rdd.toDF(["word"])

    # 注册成表
    df.createTempView("words")

    # df.printSchema()
    # df.show(truncate=False)

    # 使用SQL风格
    spark.sql("""
        select word,count(*) as cnt from words group by word order by cnt desc
    """).show()

    # TODO 2: 使用DSL风格处理WordCount
    df = spark.read.format("text").load("hdfs://node1:8020/input/wordcount/words.txt")
    # 默认列名为value

    df2 = df.select(F.explode(
        F.split(df['value'], ' ')
    ).alias('word'))
    # df['value']: 获取Column对象
    # F.split(): sql中的split操作，拆分字符串
    # F.explode(): sql中的explode方法，将array炸开，一列转多行
    # alias(): sql中的as操作，针对Column对象取别名

    df2.groupBy('word') \
        .count() \
        .withColumnRenamed('count', 'cnt') \
        .orderBy('cnt', ascending=False) \
        .show()
    # withColumnRenamed: 是针对DataFrame对象的操作，将某一列重命名
